#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <pthread.h>
#include "ringqueue.h"
#include "log.h"

#define INIT_CAPACITY   8

struct rqueue
{
    char **buf;
    int capacity, max_capacity;
    int size;
    int head, tail;
    pthread_mutex_t locker;
    pthread_cond_t not_empty, not_full;
};

enum
{
    ERR_INV_PARAM = 1, // 非法参数
    ERR_NO_MEM,        // 内存不足
    ERR_FULL,          // buffer满时，put失败
    ERR_EMPTY          // buffer空时，get失败
};

char *rqueue_strerror(int err)
{
    switch (err) {
    case 0:
        return "success";
    case ERR_INV_PARAM:
        return "invalid param";
    case ERR_NO_MEM:
        return "no memory";
    case ERR_FULL:
        return "buffer full";
    case ERR_EMPTY:
        return "buffer empty";
    default:
        return "unknown error";
    }
}

rqueue_t *rqueue_init(size_t max_elem)
{
    rqueue_t *rq = NULL;
	pthread_condattr_t attr;

    rq = (rqueue_t *)calloc(1, sizeof(rqueue_t));
    if (!rq)
        return NULL;

    rq->capacity = INIT_CAPACITY;
    rq->buf = calloc(rq->capacity, sizeof(void *));
    if (!rq->buf) {
        free(rq);
        return NULL;
    }

    if (max_elem < INIT_CAPACITY)
        max_elem = INIT_CAPACITY;
    rq->max_capacity = max_elem;
    rq->head = rq->tail = 0;
    rq->size = 0;
    pthread_mutex_init(&rq->locker, NULL);
	pthread_condattr_init(&attr);
	pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
    pthread_cond_init(&rq->not_empty, &attr);
    pthread_cond_init(&rq->not_full, &attr);
	pthread_condattr_destroy(&attr);

    return rq;
}

/*
 * 扩展内部buffer
 */
int expand_buf(rqueue_t *rq)
{
    // 空闲少于0.5时，扩容
    if (rq->size * 2 > rq->capacity && rq->capacity <= rq->max_capacity / 2) {
        int new_cap = rq->capacity * 2;
        void *p = realloc(rq->buf, new_cap * sizeof(void *));
        if (!p) {
            log_warn("failed to expand buffer capacity to %d\n", new_cap);
            return ERR_NO_MEM;
        }
        rq->buf = p;

        // 有效内容移到新空间头部
        if (rq->tail > rq->head) { // 有效内容为连续整块时，
            memmove(rq->buf,
                    &rq->buf[rq->head],
                    rq->size * sizeof(void *));
        }
        else { // 有效内容分成头块和尾块时
            int tail_blk_size = rq->capacity - rq->head, head_blk_size = rq->size - tail_blk_size;

            // 头块移到尾块后面, 拼成整块
            memmove(&rq->buf[rq->capacity],
                    rq->buf,
                    head_blk_size * sizeof(void *));

            // 整体移到头部
            memmove(rq->buf,
                    &rq->buf[rq->head],
                    rq->size * sizeof(void *));
        }

        rq->head = 0;
        rq->tail = rq->size;
        rq->capacity = new_cap;
        log_info("buffer capacity currently expand to %d\n", new_cap);
    }

    return 0;
}

static int rqueue_put_inner(rqueue_t *rq, const void *elem_data, int block, int max_wait_millisec)
{
    if (!rq)
        return ERR_INV_PARAM;

    pthread_mutex_lock(&rq->locker);
    if (!block || max_wait_millisec == 0) {
        // 由于自动扩容的原因，不可能满 !!
        if (rq->size == rq->capacity) {
            pthread_mutex_unlock(&rq->locker);
            log_info("buffer full\n");
            return ERR_FULL;
        }
    }
    else {
        if (max_wait_millisec > 0) {
            struct timespec tp;
            clock_gettime(CLOCK_MONOTONIC, &tp);
            tp.tv_sec += max_wait_millisec / 1000;
            tp.tv_nsec += max_wait_millisec % 1000 * 1000000;
            while (rq->size == rq->capacity) { // 现在满，等待
                if (pthread_cond_timedwait(&rq->not_full, &rq->locker, &tp) != 0) {
                    pthread_mutex_unlock(&rq->locker);
                    return ERR_FULL;
                }
            }
        }
        else {
            while (rq->size == rq->capacity) {
                pthread_cond_wait(&rq->not_full, &rq->locker);
            }
        }
    }

    rq->buf[rq->tail] = (char *)elem_data;
    if (rq->size == 0) // 原来是empty，放入一个元素，可唤醒
        pthread_cond_broadcast(&rq->not_empty);
    rq->tail = (rq->tail + 1) % rq->capacity;
    rq->size += 1;
    log_debug("rqueue_put(): size:%d, head:%d, tail:%d\n", rq->size, rq->head, rq->tail);
    // 检查是否需要扩容
    expand_buf(rq);
    pthread_mutex_unlock(&rq->locker);

    return 0;
}
int rqueue_put(rqueue_t *rq, const void *data, int max_wait_millisec)
{
    return rqueue_put_inner(rq, data, 1, max_wait_millisec);
}

int rqueue_try_put(rqueue_t *rq, const void *data)
{
    return rqueue_put_inner(rq, data, 0, 0);
}

static void *rqueue_get_inner(rqueue_t *rq, int block, int max_wait_millisec)
{
    void *data = NULL;

    if (!rq) {
        return NULL;
    }

    pthread_mutex_lock(&rq->locker);
    if (!block || max_wait_millisec == 0) {
        if (rq->size == 0) {
            pthread_mutex_unlock(&rq->locker);
            log_warn("buffer empty\n");
            return NULL;
        }
    }
    else {
        if (max_wait_millisec > 0) {
            struct timespec tp;
            clock_gettime(CLOCK_MONOTONIC, &tp);
            tp.tv_sec += max_wait_millisec / 1000;
            tp.tv_nsec = max_wait_millisec % 1000 * 1000000;
            while (rq->size == 0) {
                if (pthread_cond_timedwait(&rq->not_empty, &rq->locker, &tp) != 0) {
                    pthread_mutex_unlock(&rq->locker);
                    return NULL;
                }
            }
        }
        else {
            while (rq->size == 0) {
                pthread_cond_wait(&rq->not_empty, &rq->locker);
            }
        }
    }

    data = rq->buf[rq->head];
    if (rq->size == rq->capacity)
        pthread_cond_broadcast(&rq->not_full);
    rq->head = (rq->head + 1) % rq->capacity;
    rq->size -= 1;
    log_debug("rqueue_get(): size:%d, head:%d, tail:%d\n", rq->size, rq->head, rq->tail);
    pthread_mutex_unlock(&rq->locker);
    return data;
}

void *rqueue_get(rqueue_t *rq, int max_wait_millisec)
{
    return rqueue_get_inner(rq, 1, max_wait_millisec);
}

void *rqueue_try_get(rqueue_t *rq)
{
    return rqueue_get_inner(rq, 0, 0);
}

void rqueue_destroy(rqueue_t **rqp, int free_elem)
{
    rqueue_t *rq = *rqp;

    /*
    if (free_elem) {
      int i;
      for (i=0; i<rq->capacity; i++) {
        if (rq->buf[i])
          free(rq->buf[i]);
      }
    } */

    pthread_mutex_destroy(&rq->locker);
    pthread_cond_destroy(&rq->not_full);
    pthread_cond_destroy(&rq->not_empty);
    free(rq->buf);
    free(rq);
    *rqp = NULL;
}

int rqueue_size(rqueue_t *rq)
{
       return rq->size;
}

